-module(mnesia_subscribe).

-behaviour(gen_server).

-include_lib("stdlib/include/qlc.hrl").

-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-export([subscribe/2,
         unsubscribe/2,
         get_subscribers/1,
         first_run/0,
         stop/0,
         start_link/0]).
-record(subscription, {subscriber, channel}).
-record(state, {}). % state is all in mnesia
-define(SERVER, global:whereis_name(?MODULE)).

start_link() ->
    gen_server:start_link({global, ?MODULE}, ?MODULE, [], []).

stop() ->
    gen_server:call(?SERVER, {stop}).

subscribe(Subscriber, Channel) ->
    gen_server:call(?SERVER, {subscribe, Subscriber, Channel}, infinity).

unsubscribe(Subscriber, Channel) ->
    gen_server:call(?SERVER, {unsubscribe, Subscriber, Channel}, infinity).

get_subscribers(Channel) ->
    gen_server:call(?SERVER, {get_subscribers, Channel}).

%%

init([]) ->
    ok = mnesia:start(),
    %io:format("Waiting on mnesia tables..\n",[]),
    mnesia:wait_for_tables([subscription], 10000),  % 等待去连接表，超时报错
    %Info = mnesia:table_info(subscription, all),
    %io:format("OK. Subscription table info: \n~w\n\n",[Info]),
    {ok, #state{} }.

handle_call({stop}, _From, State) ->
    {stop, stop, State};

handle_call({subscribe, Subscriber, Channel}, _From, State) ->
    F = fun() -> mnesia:write(#subscription{subscriber=Subscriber,channel=Channel}) end,
    mnesia:transaction(F),
    {reply, ok, State};

handle_call({unsubscribe, Subscriber, Channel}, _From, State) ->
    List = [#subscription{subscriber=Subscriber,channel=Channel}],
    F = fun() ->
        [ ok = mnesia:delete_object(S) || S <- List ]
    end,
    mnesia:transaction(F),
    {reply, ok, State};

%%获取一个频道的所有订阅者
handle_call({get_subscribers, Channel}, From, State) ->
    Q = qlc:q([X || X <- mnesia:table(subscription), X#subscription.channel =:= Channel]),
    F = fun() -> qlc:e(Q) end,
    {atomic, Result} = mnesia:transaction(F),
    Subscribers = [Dude || #subscription{subscriber=Dude, channel=_} <- Result],
    gen_server:reply(From, Subscribers),
    {noreply, State}.

handle_cast(_Msg, State) -> {noreply, State}.
handle_info(_Msg, State) -> {noreply, State}.

terminate(_Reason, _State) ->
    mnesia:stop(),
    ok.

code_change(_OldVersion, State, _Extra) ->
    io:format("Reloading code for ?MODULE\n",[]),
    {ok, State}.

first_run() ->
    mnesia:create_schema([node()]),
    ok = mnesia:start(),
    Ret = mnesia:create_table(subscription,
    [
     {disc_copies, [node()]},
     {attributes, record_info(fields, subscription)},
     {index, [channel]}, %index subscribee too
     {type, bag}
    ]),
    Ret.
%%
